package com.dream.net.socket.base;

import com.alibaba.fastjson.JSONObject;
import com.dream.common.constans.RedisKeysDefine;
import com.dream.common.utils.JsonUtils;
import com.dream.common.utils.ThreadPoolService;
import com.dream.dto.socket.SocketBean;
import com.dream.net.socket.SocketUserChannel;
import com.dream.service.basic.RedisCacheService;
import com.dream.web.websocket.ConsoleWebSocket;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Resource;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;

import static java.util.Comparator.comparing;

public class BaseSocketController extends SimpleChannelInboundHandler {

    private Logger logger = LoggerFactory.getLogger(BaseSocketController.class);

    @Resource private ConsoleWebSocket consoleWebSocket;
    @Resource private RedisCacheService redisCacheService;
    @Resource private SocketController socketController;
    /**
     * <账户名,连接>
     *web端可以允许相同账户多个连接，app端处理新连接覆盖旧连接不处理
     */
    public static Map<String,SocketUserChannel> socketUserChannels = new ConcurrentHashMap<>(500);


    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) {
    }
    /**
     * 功能：收到连接请求
     */
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        super.handlerAdded(ctx);
        logger.info("收到连接请求! {}",ctx.name());
    }

    /**
     * 功能：获取客户端发送的消息，分发给协议处理器，获得返回的消息，然后返回给客户端
     * @param ctx 渠道处理器上下文
     * @param msg 数据
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        logger.info("channelRead 请求收到 : {}",msg);
        SocketBean socketBean = JsonUtils.parseObject((String)msg, SocketBean.class);
        //token建立,客户端发起握手
        assert socketBean != null;

        //app上线,建立socket连接
        if(socketBean.command == 100){
            JSONObject jsonObject = JSONObject.parseObject(socketBean.data);
            String account = jsonObject.getString("userName");
            //建立session,新覆盖旧连接,所以web端只会给最新的连接推，老连接不处理
            SocketUserChannel socketUserChannel = new SocketUserChannel();
            socketUserChannels.put(account,socketUserChannel);

            socketUserChannel.ctx = ctx;
            socketUserChannel.account = account;
            socketUserChannels.put(account,socketUserChannel);
            //回执,收到建立成功
            ctx.writeAndFlush(msg);

            redisCacheService.set(RedisKeysDefine.APP_USER_INFO.concat(account),socketBean.data);
            consoleWebSocket.sendMessageToUser(account,(String) msg);
            logger.info("{} : 建立socketUserChannels 缓存成功",account);
            return;
        }
        socketController.commandReceived(socketBean,(String)msg);
    }

    /*
     * 功能：异常处理
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        SocketUserChannel socketUserChannel = getSocketUserChannelByCtx(ctx);
        String yiYouId = socketUserChannel == null ? "未知ID" :socketUserChannel.account;
        logger.info("exceptionCaught 出现异常 : {}",yiYouId);
    }

    /**
     * 功能：连接不再活跃 (TCP探测包检测掉线后会回调这个 , 在此回调处理)
     * @param ctx ChannelHandlerContext
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        SocketUserChannel socketUserChannel = getSocketUserChannelByCtx(ctx);
        if(socketUserChannel == null){
            logger.info("socketUserChannel null > 未知ID");
            return;
        }

        if(isLastLinkChannel(socketUserChannel)){
            SocketBean socketBean = new SocketBean();
            socketBean.command = 101;
            socketBean.data = "offline";
            consoleWebSocket.sendMessageToUser(socketUserChannel.account,JsonUtils.toJSONString(socketBean));
        }

        logger.info("连接移除 > {}",socketUserChannel.account);
        socketUserChannel.ctx.close();
        socketUserChannels.remove(socketUserChannel.account);
    }


    /**
     * 根据ctx获取Channel
     */
    public SocketUserChannel getSocketUserChannelByCtx(ChannelHandlerContext ctx){
        for(SocketUserChannel socketUserChannels : socketUserChannels.values()){
                if(socketUserChannels.ctx.equals(ctx)){
                    return socketUserChannels;
                }
        }
        return null;
    }


    /**
     * 当前即将下线的连接是否是改用户登录的最新的一个连接
     * @return
     */
    public boolean isLastLinkChannel(SocketUserChannel userChannel){
        Optional<SocketUserChannel> optional = socketUserChannels.values().parallelStream().max(comparing(e->e.linkTime));
        return optional.map(socketUserChannel -> socketUserChannel.equals(userChannel)).orElse(false);
    }

    /**
     * 发送消息给app
     */
    public boolean sendCommand(String account, SocketBean socketBean){
        SocketUserChannel socketUserChannelByYiYouId = socketUserChannels.get(account);
        if(socketUserChannelByYiYouId == null){
            logger.info("此用户未建立连接 > {}",account);
            return false;
        }
        ThreadPoolService.execute(()->{
            String msg = JsonUtils.toJSONString(socketBean);
            socketUserChannelByYiYouId.ctx.writeAndFlush(msg + "\n");
            logger.info("发送指令成功 : {}",msg);
        });
        return true;
    }


    public void sendCommandEveryOne(SocketBean socketBean) throws RuntimeException{
        String msg = JsonUtils.toJSONString(socketBean);
        //并行处理
        socketUserChannels.values().parallelStream().forEach(e->{
            e.ctx.writeAndFlush(msg + "\n");
            logger.info("全员发送中: {} 发送指令成功 : {}",e.account,msg);
        });
    }
}
